package com.rxjava;

import java.util.concurrent.TimeUnit;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;

//过滤器操作符
public class FilterDemo {

	public static void main(String[] args) {
		//testDebounce();
		//testDistinct();
		//testElementAt();
		//testFilter();
		//testFirst();
		//testLast();
		//testIgnoreElement();
		//testSample();
		//testSkip();
		testTake();
	}

	//和skip对应，skip是跳过，take是取出
	private static void testTake() {
		Observable.just(9, 2, 3, 2, 3).take(2).subscribe(x -> System.out.println(x));
		Observable.just(9, 2, 3, 2, 3).takeLast(2).subscribe(x -> System.out.println(x));
	}

	//跳过
	private static void testSkip() {
		Observable.just(9, 2, 3, 2, 3).skip(2).subscribe(x -> System.out.println(x));
		Observable.just(9, 2, 3, 2, 3).skipLast(2).subscribe(x -> System.out.println(x));
	}

	//采样
	private static void testSample() {
		Observable.create(new ObservableOnSubscribe<Integer>() {
			@Override
			public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
				for (int i = 0; i < 10; i++) {
					Thread.sleep(1000);
					emitter.onNext(i);
				}
				emitter.onComplete();
			}
		}).sample(4, TimeUnit.SECONDS).subscribe(x -> System.out.println(x));
	}

	//ignoreElements操作符忽略所有源Observable产生的结果，只会执行onCpmpleted()或者onError()方法
	private static void testIgnoreElement() {
		Observable.just(1,2,3,4,5,6,7,8).ignoreElements()
			.subscribe(() -> System.out.println("Sequence complete."), error -> System.err.println("Error: " + error.getMessage()));
	}

	private static void testLast() {
		Observable.just(9, 2, 3, 2, 3).lastElement().subscribe(x -> System.out.println(x));
	}

	private static void testFirst() {
		Observable.just(9, 2, 3, 2, 3).firstElement().subscribe(x -> System.out.println(x));
		Observable.just(9, 2, 3, 2, 3).filter(x -> x < 4).subscribe(x -> System.out.println(x));
	}

	private static void testFilter() {
		Observable.just(1, 2, 3, 2, 3).distinct().filter(x -> x > 2).subscribe(x -> System.out.println(x));
	}

	//打印指定位数的发射项
	private static void testElementAt() {
		Observable.just(1, 2, 3, 2, 3).elementAt(2).subscribe(x -> System.out.println(x));
	}

	private static void testDistinct() {
		Observable.just(1, 2, 3, 2, 3).distinct().subscribe(x -> System.out.println(x));
	}

	//debounce：操作符是对源Observable间隔期产生的结果进行过滤，如果在这个规定的间隔期内没有别的结果产生，则将这个结果提交给订阅者，否则忽略该结果，原理有点像光学防抖.
	//debounce(时间，时间单位)
	private static void testDebounce() {
		Observable.create(new ObservableOnSubscribe<Integer>() {
			@Override
			public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
				for (int i = 0; i < 10; i++) {
					Thread.sleep(1000);
					emitter.onNext(i);
				}
				emitter.onComplete();
			}
		}).debounce(5, TimeUnit.SECONDS).subscribe(x -> System.out.println(x));
	}

}
